package org.nbict.iot.task.seriestime;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.trident.operation.BaseFilter;
import org.apache.storm.trident.operation.FlatMapFunction;
import org.apache.storm.trident.operation.MapFunction;
import org.apache.storm.trident.tuple.TridentTuple;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.nbict.iot.task.realtime.filter.UploadFilter;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.trident.influx.mapper.InfluxdbMapper;
import org.nbict.iot.trident.influx.state.InfluxdbStateFactory;
import org.nbict.iot.trident.influx.state.InfluxdbStateUpdater;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;

/**
 * 时序数据,分车辆行驶和车辆充电两种状态
 * Created by songseven on 18/7/3.
 */
public class SeriestimeExactlyTrident {

    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {
        TridentTopology topology = new TridentTopology();
        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple", // bootstrap server,192.168.1.83:9092
                        "iot-car-group-seriestime-update",  KafkaSpoutConfig.FirstPollOffsetStrategy.EARLIEST); // group id ignored
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools
                .newKafkaTridentSpoutOpaque(spoutConfig);

        Stream stream = topology.newStream("iot-car-parsed-4-seriestime", kafkaSpout);

        Fields packetFields = new Fields("car_packet_header",
                "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp",
                "upload");
        Stream stage = stream.each(new Fields("platform_id", "packet"), new
                PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header",
                        "car_packet_body"), new PacketResolver(), outputFields)
                .project(new Fields("platform_id", "car_vin", "action",
                        "timestamp", "upload"));

        Stream uploadStream = stage.each(outputFields, new UploadFilter())
                .partitionBy(new Fields("car_vin"))
                .toStream(); //2 or 3
        Stream infoStream = uploadStream.flatMap(new SeriestimeResolver(),
                new Fields("car_vin", "timestamp", "tag", "info", "isRun", "gis"));

        //01 整车
        //车辆行驶流
        Stream runStream = infoStream.filter(new BaseFilter() {
            @Override
            public boolean isKeep(TridentTuple tuple) {
                String tag = tuple.getString(2);
                boolean isRun = tuple.getBoolean(4);
                return tag.equals("01") && isRun;
            }
        });
        //车辆充电流
        Stream chargedStream = infoStream.filter(new BaseFilter() {
            @Override
            public boolean isKeep(TridentTuple tuple) {
                String tag = tuple.getString(2);
                boolean isCharged = false;
                if (tag.equals("01")) {
                    JSONObject info = (JSONObject) tuple.get(3);
                    if (info != null && (info.getString("charged_state").equals
                            ("01") )) { //停车充电 || info.getString("charged_state").equals("02") //行驶充电
                        isCharged = true;
                    }
                }
                return isCharged;
            }
        });

        //车辆行驶时序记录
        runStream.map(new MapFunction() {
            @Override
            public Values execute(TridentTuple tuple) {
                String vin = tuple.getString(0);
                Long timestamp = tuple.getLong(1);
                JSONObject info = (JSONObject) tuple.get(3);
                JSONObject gis = (JSONObject) tuple.get(5);
                return new Values(vin, timestamp, info.getDouble("odometer"), info.getFloat("speed"), info.getBigDecimal("total_current"),
                        info.getFloat("total_voltage"), info.getInteger("soc"),
                        gis.getLong("lon"), gis.getLong("lat")); //经度和维度
            }
        }, new Fields("car_vin", "time", "odometer", "speed", "total_current", "total_voltage", "soc", "lon", "lat"))//map output
                .partitionPersist(new InfluxdbStateFactory(new InfluxdbMapper("car_run_record", //influx table
                                Arrays.asList("car_vin"), //tag
                                Arrays.asList("odometer", "speed", "total_current", "total_voltage", "soc", "lon", "lat"), //influx field
                                "time")), //timestamp
                        new Fields("car_vin", "time", "odometer", "speed", "total_current", "total_voltage", "soc", "lon", "lat"), //persist input
                        new InfluxdbStateUpdater(), new Fields());

        //车辆充电时序记录
        chargedStream.map(new MapFunction() {
            @Override
            public Values execute(TridentTuple tuple) {
                String vin = tuple.getString(0);
                Long timestamp = tuple.getLong(1);
                JSONObject info = (JSONObject) tuple.get(3);
                JSONObject gis = (JSONObject) tuple.get(5);
                return new Values(vin, timestamp, info.getInteger("soc"), gis.getLong("lon"), gis.getLong("lat"));
            }
        }, new Fields("car_vin", "time", "soc", "lon", "lat")) //map output
            .partitionPersist(new InfluxdbStateFactory(new InfluxdbMapper("car_charged_record",
                            Arrays.asList("car_vin"),
                            Arrays.asList("soc", "lon", "lat"),
                            "time")),
                    new Fields("car_vin", "time", "soc", "lon", "lat"),
                    new InfluxdbStateUpdater(), new Fields());


//        //05 定位(只记录车辆行驶状态)
//        infoStream.filter(new BaseFilter() {
//            @Override
//            public boolean isKeep(TridentTuple tuple) {
//                String tag = tuple.getString(2);
//                boolean isRun = tuple.getBoolean(4); // 是否启动状态
//                return tag.equals("05") && isRun;
//            }
//        }).map(new MapFunction() {
//            @Override
//            public Values execute(TridentTuple tuple) {
//                String vin = tuple.getString(0);
//                Long timestamp = tuple.getLong(1);
//                JSONObject info = (JSONObject) tuple.get(3);
//                return new Values(vin, timestamp, info.getLongValue("lon"), info.getLongValue("lat"));
//            }
//        }, new Fields("car_vin", "time", "lon", "lat"))
//                .partitionPersist(new InfluxdbStateFactory(new InfluxdbMapper("car_gis_record", //influx table
//                        Arrays.asList("car_vin"), //tag
//                        Arrays.asList("lon", "lat"), //influx field
//                        "time")), //timestamp
//                new Fields("car_vin", "time", "lon", "lat"), //persist input
//                new InfluxdbStateUpdater(), new Fields());

        //02 驱动电机
        infoStream.filter(new BaseFilter() {
            @Override
            public boolean isKeep(TridentTuple tuple) {
                String tag = tuple.getString(2);
                boolean isRun = tuple.getBoolean(4);
                return tag.equals("02") && isRun;
            }
        }).flatMap(new FlatMapFunction() {
            @Override
            public Iterable<Values> execute(TridentTuple tuple) {
                String vin = tuple.getString(0);
                Long timestamp = tuple.getLong(1);
                JSONObject info = (JSONObject) tuple.get(3);
                List<Values> valuesList = new ArrayList<>();
                JSONArray array = info.getJSONArray("driver_moter_list");
                for (Iterator iterator = array.iterator(); iterator.hasNext();){
                    JSONObject object = (JSONObject) iterator.next();
                    Integer temperature = object.getInteger("temp"); //电机温度
                    Integer rpm = object.getInteger("rpm"); //电机转速
                    String moterId = object.getString("driver_moter_id"); //电机序号
                    Integer ctlTemperature = object.getInteger("ctrl_temp");
                    Float ctlVoltage = object.getFloat("ctrl_input_voltage");
                    Float ctlCurrent = object.getFloat("ctr_dc_bus_current");
                    valuesList.add(new Values(vin, timestamp, moterId, temperature, rpm, ctlTemperature, ctlVoltage, ctlCurrent));
                }
                return valuesList;
            }
        }, new Fields("car_vin", "time", "moter_id", "temperature", "rpm",
                "ctl_temperature", "ctl_voltage", "ctl_current"))//outpout
                .partitionPersist(new InfluxdbStateFactory(new InfluxdbMapper
                                ("car_motor_record", //influx table
                                Arrays.asList("car_vin", "moter_id"), //tag
                                Arrays.asList("temperature", "rpm",
                                        "ctl_temperature", "ctl_voltage",
                                        "ctl_current"), //influx field
                                "time")), //timestamp
                        new Fields("car_vin", "time", "moter_id",
                                "temperature", "rpm", "ctl_temperature",
                                "ctl_voltage", "ctl_current"), //persist input
                        new InfluxdbStateUpdater(), new Fields());

        Config config = new Config();
        config.setMaxSpoutPending(3);

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-gis", config, topology.build());
        }
    }
}